import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

import scala.collection.mutable

object SparkToN {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)
    val spark = SparkSession.builder()
      .master("local")
      .appName("SparkToN")
      .getOrCreate()
    val dataRDD: RDD[String] = spark.sparkContext.textFile("log.txt")
    /**
     * 第一步：
     * 获取所有品类ID
     */
    val allCategoryIds: RDD[(Long, Long)] = getAllCategoryId(dataRDD).distinct()
    /**
     * 第二步：
     * 分别获取品类的:点击，下单，支付的次数
     */
    val clickCategoryCount: RDD[(Long, Long)] = getClickCategoryCount(dataRDD)
    val orderCategoryCount: RDD[(Long, Long)] = getOrderCategoryCount(dataRDD)
    val payCategoryCount: RDD[(Long, Long)] = getPayCategoryCount(dataRDD)
    /**
     * 第三步：
     * 第一步的结果与第二步的结果进行left join
     */
    val resultRDD: RDD[(Long, String)] =
      leftJoinRDD(allCategoryIds,clickCategoryCount,orderCategoryCount,payCategoryCount)
    /**
     * 第四步：
     * 实现二次排序的效果
     */
    getTopN(resultRDD)
    spark.close()
  }
  /**
   * 获取所有数据的品类ID
   * @param dataRDD
   * @return
   */
  def getAllCategoryId(dataRDD: RDD[String]):RDD[(Long,Long)]={
    val ids = new mutable.HashSet[(Long,Long)]
    dataRDD.flatMap(line =>{
      val fields = line.split(",")
      val click_category_id = fields(7)
      val order_category_id = fields(9)
      val pay_category_id = fields(11)
      if(click_category_id !=null && !click_category_id.equals("")){
        ids += ((click_category_id.toLong,click_category_id.toLong))
      }
      if(order_category_id !=null && !order_category_id.equals("")){
        val fields = order_category_id.split("\\^A")
        for(categoryid <- fields){
          ids+=((categoryid.toLong,categoryid.toLong))
        }
      }
      if(pay_category_id !=null && !pay_category_id.equals("")){
        val fields = pay_category_id.split("\\^A")
        for(categoryid <- fields){
          ids+=((categoryid.toLong,categoryid.toLong))
        }
      }
      ids
    })
  }
  /**
   * 计算品类点击的次数
   */
  def getClickCategoryCount(dataRDD: RDD[String]):RDD[(Long,Long)]={
    dataRDD.filter( line =>{
      val fields = line.split(",")
      fields(7) != null && !fields.equals("")
    }).map( line =>{
      val clickCategoryId = line.split(",")(7).toLong
      (clickCategoryId,1L)
    }).reduceByKey(_+_)
  }
  /**
   * 获取品类下单的次数
   */
  def getOrderCategoryCount(dataRDD: RDD[String]):RDD[(Long,Long)]={
    dataRDD.filter( line =>{
      val fields = line.split(",")
      fields(9) != null && !fields.equals("")
    }).flatMap( line =>{
      line.split(",")(9).split("\\^A")
    }.map( orderCategoryId =>{
      (orderCategoryId.toLong,1L)
    })).reduceByKey(_+_)
  }
  /**
   * 获取品类支付的次数
   * @param dataRDD
   * @return
   */
  def getPayCategoryCount(dataRDD: RDD[String]):RDD[(Long,Long)]={
    dataRDD.filter( line =>{
      val fields = line.split(",")
      fields(11) != null && !fields.equals("")
    }).flatMap( line =>{
      line.split(",")(11).split("\\^A")
    }.map( payCategoryId =>{
      (payCategoryId.toLong,1L)
    })).reduceByKey(_+_)
  }
  /**
   *
   * 1
category_id=1|click_category_count=2|order_category_count=5|pay_category_count=5
   * 全品类ID与所有的点击，下单，支付的品类进行join
   */
  def leftJoinRDD(allCategoryIds: RDD[(Long, Long)],
                  clickCategoryCount: RDD[(Long, Long)] ,
                  orderCategoryCount: RDD[(Long, Long)],
                  payCategoryCount: RDD[(Long, Long)]
                 ):RDD[(Long,String)]={
    /**
     * Option:
     * Some:有值
     * None：没有值
     * Long：品类ID
     * Long：品类ID
     * Option：品类被点击的次数
     */
    val resultRDD: RDD[(Long, (Long, Option[Long]))] =
      allCategoryIds.leftOuterJoin(clickCategoryCount)
    /**
     * Long: 品类
     * String： 上一次的拼接的结果
     * Option：下单品类的次数
     */
    val result2RDD: RDD[(Long, (String, Option[Long]))] = resultRDD.map(tuple => {
      val categroy_id: Long = tuple._1
      val clickCategorCount = tuple._2._2.getOrElse(0)
      //注意，在公司里面这儿必要出现硬编码
      val value = "category_id=" + categroy_id + "|click_category_count=" +
        clickCategorCount
      (categroy_id, value)
    }).leftOuterJoin(orderCategoryCount)
    /**
     * Long: 品类
     * String： 上一次的拼接的结果
     * Option：支付品类的次数
     */
    val result3RDD: RDD[(Long, (String, Option[Long]))] = result2RDD.map(tuple =>
    {
      val category_id = tuple._1
      var value = tuple._2._1
      val count = tuple._2._2.getOrElse(0)
      value += "|order_category_count=" + count
      (category_id, value)
    }).leftOuterJoin(payCategoryCount)
    result3RDD.map( tuple =>{
      val category_id = tuple._1
      var value = tuple._2._1
      val count = tuple._2._2.getOrElse(0)
      value += "|pay_category_count="+ count
      (category_id, value)
    })
  }
  def getTopN(resultRDD:RDD[(Long,String)])={
    resultRDD.map( tuple =>{
      // val category_id = tuple._1
      val value = tuple._2
      //category_id=4|click_category_count=1|order_category_count=0|pay_category_count=
      1
      val click_count = value.split("\\|")(1).split("=")(1).toLong
      val order_count = value.split("\\|")(2).split("=")(1).toLong
      val pay_count = value.split("\\|")(3).split("=")(1).toLong
      val key = new SortKey(click_count,order_count,pay_count)
      (key,value)
    }).sortByKey(false)
      .foreach( tuple =>{
        println(tuple._2)
      })
  }
}
//排序类
class SortKey(var clickCount:Long,
              var orderCount:Long,
              var payCount:Long) extends Ordered[SortKey] with Serializable{
  override def compare(that: SortKey): Int = {
    if(this.clickCount - that.clickCount != 0){
      if(this.clickCount - that.clickCount < 0){
        -1
      }else{
        1
      }
    }else if(this.orderCount - that.orderCount != 0){
      if(this.orderCount - that.orderCount <0){
        -1
      }else{
        1
      }
    }else{
      if(this.payCount - that.payCount <0){
        -1
      }else{
        1
      }
    }
  }
}